﻿using MQTTnet.Client;
using MQTTnet;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using MqttApplicationMessage = MQTTnet.MqttApplicationMessage;
using MQTTnet.Protocol;
using System.Threading;
using MQTTnet.Server;
using System.Configuration;
using System.Diagnostics;
using FlowerPot;
using Newtonsoft.Json;
using static System.Collections.Specialized.BitVector32;
using System.Data;
using System.IO;
using Newtonsoft.Json.Linq;

namespace HuanReZhanDataReceiver
{
    public class MqttService
    {
        public static MqttServer _mqttServer { get; set; }

        public static void PublishData(string data)
        {
            var message = new MqttApplicationMessage
            {
                Topic = "topic1",
                Payload = Encoding.Default.GetBytes(data),
                QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce,
                Retain = true  // 服务端是否保留消息。true为保留，如果有新的订阅者连接，就会立马收到该消息。
            };

            _mqttServer.InjectApplicationMessage(new InjectedMqttApplicationMessage(message) // 发送消息给有订阅 topic_01的客户端
            {
                SenderClientId = "Server_01"
            }).GetAwaiter().GetResult();
        }

    }

    public class MqttHostService :IDisposable
    {
        public void Dispose()
        {

        }
        const string ServerClientId = "SERVER";
        public Task StartAsync(CancellationToken cancellationToken)
        {
            MqttServerOptionsBuilder optionsBuilder = new MqttServerOptionsBuilder();
            optionsBuilder.WithDefaultEndpoint();
            //optionsBuilder.WithDefaultEndpointBoundIPAddress(IPAddress.Parse("127.0.0.1"));
            optionsBuilder.WithDefaultEndpointPort(10086); // 设置 服务端 端口号
            optionsBuilder.WithConnectionBacklog(1000); // 最大连接数
            MqttServerOptions options = optionsBuilder.Build();

            MqttService._mqttServer = new MqttFactory().CreateMqttServer(options);

            MqttService._mqttServer.ClientConnectedAsync += _mqttServer_ClientConnectedAsync; //客户端连接事件
            MqttService._mqttServer.ClientDisconnectedAsync += _mqttServer_ClientDisconnectedAsync; // 客户端关闭事件
            MqttService._mqttServer.ApplicationMessageNotConsumedAsync += _mqttServer_ApplicationMessageNotConsumedAsync; // 消息接收事件

            MqttService._mqttServer.ClientSubscribedTopicAsync += _mqttServer_ClientSubscribedTopicAsync; // 客户端订阅主题事件
            MqttService._mqttServer.ClientUnsubscribedTopicAsync += _mqttServer_ClientUnsubscribedTopicAsync; // 客户端取消订阅事件
            MqttService._mqttServer.StartedAsync += _mqttServer_StartedAsync; // 启动后事件
            MqttService._mqttServer.StoppedAsync += _mqttServer_StoppedAsync; // 关闭后事件
            MqttService._mqttServer.InterceptingPublishAsync += _mqttServer_InterceptingPublishAsync; // 消息接收事件
            MqttService._mqttServer.ValidatingConnectionAsync += _mqttServer_ValidatingConnectionAsync; // 用户名和密码验证有关

            MqttService._mqttServer.StartAsync();
            return Task.CompletedTask;
        }

        /// <summary>
        /// 客户端订阅主题事件
        /// </summary>
        /// <param name="arg"></param>
        /// <returns></returns>
        private Task _mqttServer_ClientSubscribedTopicAsync(ClientSubscribedTopicEventArgs arg)
        {
            Console.WriteLine($"ClientSubscribedTopicAsync：客户端ID=【{arg.ClientId}】订阅的主题=【{arg.TopicFilter}】 ");
            return Task.CompletedTask;
        }

        /// <summary>
        /// 关闭后事件
        /// </summary>
        /// <param name="arg"></param>
        /// <returns></returns>
        private Task _mqttServer_StoppedAsync(EventArgs arg)
        {
            Console.WriteLine($"StoppedAsync：MQTT服务已关闭……");
            return Task.CompletedTask;
        }

        /// <summary>
        /// 用户名和密码验证有关
        /// </summary>
        /// <param name="arg"></param>
        /// <returns></returns>
        private Task _mqttServer_ValidatingConnectionAsync(ValidatingConnectionEventArgs arg)
        {
            arg.ReasonCode = MqttConnectReasonCode.Success;
            if ((arg.Username ?? string.Empty) != "admin" || (arg.Password ?? String.Empty) != "123456")
            {
                arg.ReasonCode = MqttConnectReasonCode.Banned;
                Console.WriteLine($"ValidatingConnectionAsync：客户端ID=【{arg.ClientId}】用户名或密码验证错误 ");

            }
            return Task.CompletedTask;
        }

        /// <summary>
        /// 消息接收事件
        /// </summary>
        /// <param name="arg"></param>
        /// <returns></returns>
        private Task _mqttServer_InterceptingPublishAsync(InterceptingPublishEventArgs arg)
        {
            if (string.Equals(arg.ClientId, ServerClientId))
            {
                return Task.CompletedTask;
            }

            Console.WriteLine($"InterceptingPublishAsync：客户端ID=【{arg.ClientId}】 Topic主题=【{arg.ApplicationMessage.Topic}】 消息=【{Encoding.UTF8.GetString(arg.ApplicationMessage.Payload)}】 qos等级=【{arg.ApplicationMessage.QualityOfServiceLevel}】");
            return Task.CompletedTask;

        }

        /// <summary>
        /// 启动后事件
        /// </summary>
        /// <param name="arg"></param>
        /// <returns></returns>
        private Task _mqttServer_StartedAsync(EventArgs arg)
        {
            Console.WriteLine($"StartedAsync：MQTT服务已启动……");
            return Task.CompletedTask;
        }

        /// <summary>
        /// 客户端取消订阅事件
        /// </summary>
        /// <param name="arg"></param>
        /// <returns></returns>
        private Task _mqttServer_ClientUnsubscribedTopicAsync(ClientUnsubscribedTopicEventArgs arg)
        {
            Console.WriteLine($"ClientUnsubscribedTopicAsync：客户端ID=【{arg.ClientId}】已取消订阅的主题=【{arg.TopicFilter}】  ");
            return Task.CompletedTask;
        }

        private Task _mqttServer_ApplicationMessageNotConsumedAsync(ApplicationMessageNotConsumedEventArgs arg)
        {
            Console.WriteLine($"ApplicationMessageNotConsumedAsync：发送端ID=【{arg.SenderId}】 Topic主题=【{arg.ApplicationMessage.Topic}】 消息=【{Encoding.UTF8.GetString(arg.ApplicationMessage.Payload)}】 qos等级=【{arg.ApplicationMessage.QualityOfServiceLevel}】");
            return Task.CompletedTask;

        }

        /// <summary>
        /// 客户端断开时候触发
        /// </summary>
        /// <param name="arg"></param>
        /// <returns></returns>
        /// <exception cref="NotImplementedException"></exception>
        private Task _mqttServer_ClientDisconnectedAsync(ClientDisconnectedEventArgs arg)
        {
            Console.WriteLine($"ClientDisconnectedAsync：客户端ID=【{arg.ClientId}】已断开, 地址=【{arg.Endpoint}】  ");
            return Task.CompletedTask;

        }

        /// <summary>
        /// 客户端连接时候触发
        /// </summary>
        /// <param name="arg"></param>
        /// <returns></returns>
        private Task _mqttServer_ClientConnectedAsync(ClientConnectedEventArgs arg)
        {
            Console.WriteLine($"ClientConnectedAsync：客户端ID=【{arg.ClientId}】已连接, 用户名=【{arg.UserName}】地址=【{arg.Endpoint}】  ");
            return Task.CompletedTask;
        }

        public Task StopAsync(CancellationToken cancellationToken)
        {
            return Task.CompletedTask;
        }
    }

    public class MqttClientService
    {
        public static IMqttClient _mqttClient;
        public void MqttClientStart()
        {
            var optionsBuilder = new MqttClientOptionsBuilder()
                .WithTcpServer("101.201.68.169", 6071) // 要访问的mqtt服务端的 ip 和 端口号
                .WithCredentials("admin", "public") // 要访问的mqtt服务端的用户名和密码
                .WithClientId(ConfigurationManager.AppSettings["MQTTClientVersion"]) // 设置客户端id
                .WithCleanSession()
                .WithTls(new MqttClientOptionsBuilderTlsParameters
                {
                    UseTls = false  // 是否使用 tls加密
                });

            var clientOptions = optionsBuilder.Build();
            _mqttClient = new MqttFactory().CreateMqttClient();

            _mqttClient.ConnectedAsync += _mqttClient_ConnectedAsync; // 客户端连接成功事件
            _mqttClient.DisconnectedAsync += _mqttClient_DisconnectedAsync; // 客户端连接关闭事件
            _mqttClient.ApplicationMessageReceivedAsync += _mqttClient_ApplicationMessageReceivedAsync; // 收到消息事件

            _mqttClient.ConnectAsync(clientOptions);


        }

        public void MqttClientStop()
        {
            if (_mqttClient.IsConnected)
            {
                _mqttClient.DisconnectAsync();
            }
        }

        /// <summary>
        /// 客户端连接关闭事件
        /// </summary>
        /// <param name="arg"></param>
        /// <returns></returns>
        private Task _mqttClient_DisconnectedAsync(MqttClientDisconnectedEventArgs arg)
        {
            Trace.WriteLine($"客户端已断开与服务端的连接……");
            Form1.showLogs($"客户端已断开与服务端的连接……");
            return Task.CompletedTask;
        }

        /// <summary>
        /// 客户端连接成功事件
        /// </summary>
        /// <param name="arg"></param>
        /// <returns></returns>
        private Task _mqttClient_ConnectedAsync(MqttClientConnectedEventArgs arg)
        {
            Trace.WriteLine($"客户端已连接服务端……");
            Form1.showLogs($"客户端已连接服务端……");

            // 订阅消息主题
            // MqttQualityOfServiceLevel: （QoS）:  0 最多一次，接收者不确认收到消息，并且消息不被发送者存储和重新发送提供与底层 TCP 协议相同的保证。
            // 1: 保证一条消息至少有一次会传递给接收方。发送方存储消息，直到它从接收方收到确认收到消息的数据包。一条消息可以多次发送或传递。
            // 2: 保证每条消息仅由预期的收件人接收一次。级别2是最安全和最慢的服务质量级别，保证由发送方和接收方之间的至少两个请求/响应（四次握手）。
            _mqttClient.SubscribeAsync("$dp", MqttQualityOfServiceLevel.AtLeastOnce); //$dp

            return Task.CompletedTask;
        }

        /// <summary>
        /// 收到消息事件
        /// </summary>
        /// <param name="arg"></param>
        /// <returns></returns>
        private Task _mqttClient_ApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs arg)
        {
            
            Trace.WriteLine($"ApplicationMessageReceivedAsync：客户端ID=【{arg.ClientId}】接收到消息。 Topic主题=【{arg.ApplicationMessage.Topic}】 消息=【{Encoding.UTF7.GetString(arg.ApplicationMessage.Payload)}】 qos等级=【{arg.ApplicationMessage.QualityOfServiceLevel}】");
            Form1.showLogs($"ApplicationMessageReceivedAsync：客户端ID=【{arg.ClientId}】接收到消息。 Topic主题=【{arg.ApplicationMessage.Topic}】 消息=【{Encoding.UTF7.GetString(arg.ApplicationMessage.Payload)}】 qos等级=【{arg.ApplicationMessage.QualityOfServiceLevel}】");

            Trace.WriteLine("========================================");
            operateRecData(arg.ClientId,arg.ApplicationMessage.Topic,Encoding.UTF7.GetString(arg.ApplicationMessage.Payload));

            return Task.CompletedTask;
        }

        [Obsolete]
        public void Publish(string data)
        {
            var message = new MqttApplicationMessage
            {
                Topic = "$dp",
                Payload = Encoding.Default.GetBytes(data),
                QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce,
                Retain = true  // 服务端是否保留消息。true为保留，如果有新的订阅者连接，就会立马收到该消息。
            };
            _mqttClient.PublishAsync(message);
        }


        public void operateRecData(String clientId,String topic,String dataStr)
        {
            JObject json = new JObject();
            
            json.Add("clientId", clientId);
            json.Add("topic", topic);
            json.Add("dataStr", dataStr);
            json.Add("time", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"));
            File.AppendAllText("D:\\mqtt.txt", json.ToString() + "\n");

            String[] lines = dataStr.Split('\n');
            for (int i = 0; i < lines.Length; i++)
            {
                String line = lines[i];
                if (line.Trim().Equals("")) { 
                    continue;
                }

                if (line.IndexOf("INFO") > 0)
                {
                    line = line.Substring(line.IndexOf("{"));
                    MQTTSensorData_INFO info = JsonConvert.DeserializeObject<MQTTSensorData_INFO>(line);
                    Form1.showLogs("INFO: " + info.ToString());
                }
                else if (line.IndexOf("HISTORYDATA") > 0)
                {
                    line = line.Substring(line.IndexOf("{"));
                    //MQTTSensorData_HISTORYDATA historyData = JsonConvert.DeserializeObject<MQTTSensorData_HISTORYDATA>(line);
                    MQTTSensorData_HISTORYDATA historyData = JsonConvert.DeserializeObject<MQTTSensorData_HISTORYDATA>(JsonConvert.DeserializeObject<Dictionary<String, Object>>(line)["HISTORYDATA"].ToString());

                    Form1.showLogs("HISTORYDATA: " + historyData.ToString());

                    addHistoryDataToDB(historyData);
                }
                else if (line.IndexOf("DATA") > 0)
                {
                    line = line.Substring(line.IndexOf("{"));
                    //MQTTSensorData_DATA data = JsonConvert.DeserializeObject<MQTTSensorData_DATA>(line);
                    MQTTSensorData_DATA data = JsonConvert.DeserializeObject<MQTTSensorData_DATA>(JsonConvert.DeserializeObject<Dictionary<String, Object>>(line)["DATA"].ToString());

                    Form1.showLogs("DATA: " + data.ToString());

                    addDataToDB(data);
                }
                
            }
        }

        public void addDataToDB(MQTTSensorData_DATA data) {

            String imei = data.IMEI;
            System.DateTime startTime = TimeZone.CurrentTimeZone.ToLocalTime(new System.DateTime(1970, 1, 1));//当地时区
            var time = startTime.AddSeconds(data.T_P);
            String dataValue = JsonConvert.SerializeObject(data);
            String sql1 = "INSERT INTO receive_history_new (station_id,data_id,value,created_time) VALUES('"+ imei + "',1,'"+ dataValue + "','" + time.ToString("yyyy-MM-dd HH:mm:ss") + "')";
            SqlHelper.ExecuteNonQuery(ConfigurationManager.AppSettings["ConnectString"], CommandType.Text, sql1, null);
        }

        public void addHistoryDataToDB(MQTTSensorData_HISTORYDATA data)
        {
            String imei = data.IMEI;
            System.DateTime startTime = TimeZone.CurrentTimeZone.ToLocalTime(new System.DateTime(1970, 1, 1));//当地时区
            var time = startTime.AddSeconds(data.T_P);
            String dataValue = JsonConvert.SerializeObject(data);
            String sql1 = "INSERT INTO receive_history_new (station_id,data_id,value,created_time) VALUES('" + imei + "',1,'" + dataValue + "','" + time.ToString("yyyy-MM-dd HH:mm:ss") + "')";
            SqlHelper.ExecuteNonQuery(ConfigurationManager.AppSettings["ConnectString"], CommandType.Text, sql1, null);
        }
    }
}

public class MQTTSensorData_INFO
{ 
	public String VERSION { get; set; }
    public String SN { get; set; }
    public String IMEI { get; set; }
    public String CCID { get; set; }
    public int DEV_TYPE { get; set; }
    public String NET_TYPE { get; set; }
    public int COLLECT_P { get; set; }
    public int RECORD_P { get; set; }
    public int UPLOAD_P { get; set; }

    public override String ToString()
    {
        return "VERSION: " + VERSION + ", SN: " + SN + ", IMEI: " + IMEI + ", CCID: " + CCID + ", DEV_TYPE: " + DEV_TYPE + ", NET_TYPE: " + NET_TYPE + ", COLLECT_P: " + COLLECT_P + ", RECORD_P: " + RECORD_P + ", UPLOAD_P: " + UPLOAD_P;
    }
}

public class MQTTSensorData_DATA
{
    public String IMEI { get; set; }
    public int CSQ { get; set; }
    public String BV { get; set; }
    public float RANGEH_P { get; set; }
    public float RANGEL_P { get; set; }
    public float RANGEH_T { get; set; }
    public float RANGEL_T { get; set; }
    public String UNIT_P { get; set; }
    public String UNIT_T { get; set; }
    public float HIAL_P { get; set; }
    public float LOAL_P { get; set; }
    public float WAAL_P { get; set; }
    public int ALEN_P { get; set; }
    public int ALST_P { get; set; }
    public float HIAL_T { get; set; }
    public float LOAL_T { get; set; }
    public float WAAL_T { get; set; }
    public int ALEN_T { get; set; }
    public int ALST_T { get; set; }    
    public float PRESSURE { get; set; }
    public float TEMPERATURE { get; set; }
    public long T_P { get; set; }
    public String ERRCODE { get; set; }

    public override string ToString()
    {
        return "CSQ: " + CSQ + ", BV: " + BV + ", RANGEH_P: " + RANGEH_P + ", RANGEL_P: " + RANGEL_P + ", RANGEH_T: " + RANGEH_T + ", RANGEL_T: " + RANGEL_T + ", UNIT_P: " + UNIT_P + ", UNIT_T: " + UNIT_T + ", HIAL_P: " + HIAL_P + ", LOAL_P: " + LOAL_P + ", WAAL_P: " + WAAL_P + ", ALEN_P: " + ALEN_P + ", ALST_P: " + ALST_P + ", HIAL_T: " + HIAL_T + ", LOAL_T: " + LOAL_T + ", WAAL_T: " + WAAL_T + ", ALEN_T: " + ALEN_T + ", ALST_T: " + ALST_T + ", PRESSURE: " + PRESSURE + ", TEMPERATURE: " + TEMPERATURE + ", T_P: " + T_P + ", ERRCODE: " + ERRCODE;
    }
}

public class MQTTSensorData_HISTORYDATA
{
    public String IMEI { get; set; }
    public Double PRESSURE { get; set; }
    public Double TEMPERATURE { get; set; }
    public long T_P { get; set; }
    public int ALST_P { get; set; }
    public int ALST_T { get; set; }

    public override string ToString()
    {
        return "PRESSURE: " + PRESSURE + ", TEMPERATURE: " + TEMPERATURE + ", T_P: " + T_P + ", ALST_P: " + ALST_P + ", ALST_T: " + ALST_T;
    }
}